package com.young.broadcast;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 广播模式的消费者(如果不特意的设置为为广播模式，则默认的模式为所有消费者负载均衡的消费MQ中的消息)
 * 消费者采用广播模式消费消息，每个消费者消费的消息都是想通的
 */
public class BroadcastConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("broadcast_consumer_group");
        consumer.setNamesrvAddr("192.168.188.101:9876;192.168.188.102:9876;");
        consumer.subscribe("broadcast_topic","*");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                list.forEach(message -> {
                    String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
                    System.out.println("接收到消息, topic:"+message.getTopic()+"Tags："+message.getTags()+"消息内容："+messageBody);
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("RocketMQ 广播模式消费者启动......");
    }
}
